package cn.z2huo.demo.kafka.apache.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * <p>同步发送消息的生产者
 *
 * @author z2huo
 */
@Slf4j
public class SyncProducer {

    public static void main(String[] args) {
        new SyncProducer().sendMessageSync();
    }

    public void sendMessageSync() {
        Properties props = ProducerProperties.getProducerProperties();
        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {

            String message = "Hello Kafka, from sync send message, time is " + LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME);

            ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", message);
            // 发送消息，同步发送
            RecordMetadata metadata = producer.send(record).get();
            log.info("send message to topic {}, partition {} with offset {}, value is {}", metadata.topic(), metadata.partition(), metadata.offset(), message);
            // 刷新并关闭生产者
            producer.flush();
        } catch (ExecutionException | InterruptedException e) {
            throw new RuntimeException(e);
        }

    }

}
